package yarnrpc

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable


class MyResourceManager(val hostname:String, val port : Int) extends  Actor{
  private val map = new mutable.HashMap[String, NodeManagerInfo]()
  private val set = new mutable.HashSet[NodeManagerInfo]()


  override def preStart(): Unit = {
    import scala.concurrent.duration._
    import context.dispatcher
    context.system.scheduler.schedule(0 millis, Constant.RESOURCE_MANAGER_CHECK_TIME millis, self, SendHeartBeatChecktoSelf)
  }

  def handleRegisterNodeManager(nodeManagerId:String, cpu:Int, mem:Int):Unit={
    println(s"新的NodeManager加入: nodeManagerId:${nodeManagerId} cpu:${cpu} mem:${mem}")
    val nodeManagerInfo = new NodeManagerInfo(nodeManagerId, cpu, mem, System.currentTimeMillis())
    map.put(nodeManagerId, nodeManagerInfo)
    set.add(nodeManagerInfo)
    sender() ! new RegisteredNodeManager(hostname)
  }

  def handleHeartBeat(nodeManagerId:String):Unit = {
    if(map.contains(nodeManagerId)){
      val nodeManagerInfo = map(nodeManagerId)
      nodeManagerInfo.lastHearBeatTime = System.currentTimeMillis()
      map.put(nodeManagerId, nodeManagerInfo)
      set.add(nodeManagerInfo)
    }

  }

  def handleSendHeartBeatChecktoSelf():Unit = {
      var currentTime = System.currentTimeMillis()
      for(nodeManagerInfo <- set){
        if(currentTime - nodeManagerInfo.lastHearBeatTime > Constant.NODE_MANAGER_DEAD_TIME){
          set-=nodeManagerInfo
          map.remove(nodeManagerInfo.nodeManagerId)
          println(s"移除节点: nodeManagerId:${nodeManagerInfo.nodeManagerId} cpu:${nodeManagerInfo.cpu} " +
            s"mem:${nodeManagerInfo.mem} lastHearBeatTime:${nodeManagerInfo.lastHearBeatTime}")
        }
      }
    println(s"当前活跃节点数: ${set.size}")
  }
  override def receive: Receive = {
    case RegisterNodeManager(nodeManagerId , cpu, mem) =>{
      handleRegisterNodeManager(nodeManagerId , cpu, mem)
    }
      // node manager 心跳整理
    case HeartBeat(nodeManagerId) =>{
      handleHeartBeat(nodeManagerId)
    }
    // 心跳检查
    case SendHeartBeatChecktoSelf()=>{
      handleSendHeartBeatChecktoSelf()
    }
    // 心跳检查
    case SendHeartBeatChecktoSelf=>{
      handleSendHeartBeatChecktoSelf()
    }
    case _ =>{
      println("不应该到这里")
    }
  }
}

object MyResourceManager {
  // apply 默认应用
  def apply(hostname:String, port : Int): MyResourceManager = new MyResourceManager(hostname, port)
  def main(args: Array[String]): Unit = {
    // 本机主机名
    val RESOURCEMANAGER_HOSTNAME = args(0)           // 本机主机名
    // 本机监听的端口号
    val RESOURCEMANAGER_PORT = args(1).toInt         // 本机监听的端口号

    println(s"ResourceManager  host:${RESOURCEMANAGER_HOSTNAME} port:${RESOURCEMANAGER_PORT}")

    val str=
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname =${RESOURCEMANAGER_HOSTNAME}
        |akka.remote.netty.tcp.port=${RESOURCEMANAGER_PORT}
      """.stripMargin

    val config = ConfigFactory.parseString(str)
    val myResourceManagerActorSystem = ActorSystem(Constant.RESOURCE_MANAGER_ACTOR_SYSTEM ,config)
    myResourceManagerActorSystem.actorOf(Props(MyResourceManager(RESOURCEMANAGER_HOSTNAME, RESOURCEMANAGER_PORT)), Constant.RESOURCE_MANAGER_ACTOR)
  }
}
